package com.zhang.sql2;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

/**
 * @title:
 * @author: zhang
 * @date: 2022/3/26 18:49
 */
public class MyAggregateFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2 创建表 (连接器表)
        String create = "create table clickTable (" +
                " username STRING," +
                " url STRING," +
                " ts BIGINT" +
                " ) WITH ( " +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/click.txt'," +
                " 'format' = 'csv'" +
                " )";
        tableEnv.executeSql(create);


        //注册自定义函数
        tableEnv.createTemporarySystemFunction("MyAvg",weightedAverage.class);


        //调用自定义函数
        Table table = tableEnv.sqlQuery("" +
                "select username,MyAvg(ts,1) from clickTable group by username");

        //转换成流打印输出
        tableEnv.toChangelogStream(table).print();
        env.execute();
    }
    //定义一个累加器类型
    public static class WeightedAvgAccumulator{
        public long sum = 0;
        public int count = 0;
    }


    //实现加权平均值
    //Long 返回值类型
    //WeightedAvgAccumulator 中间累加器类型
    public static class weightedAverage extends AggregateFunction<Long,WeightedAvgAccumulator>{

        @Override
        public Long getValue(WeightedAvgAccumulator accumulator) {
            if (accumulator.count == 0){
                return null;
            }else {
                return accumulator.sum/accumulator.count;
            }
        }

        @Override
        public WeightedAvgAccumulator createAccumulator() {
            return new WeightedAvgAccumulator();
        }

        //此方法自己实现，方法名字必须为accumulate，第一个参数为累加器，后面是调用函数传入的参数
        public void accumulate(WeightedAvgAccumulator accumulator,Long iVale,Integer iWeight){
            accumulator.sum += iVale * iWeight;
            accumulator.count += iWeight;
        }
    }
}
